package org.niit.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/*
 join: 将两个流 相同数据进行合并

 */
object SparkStreaming_05 {
  def main(args: Array[String]): Unit = {
     //准备环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    ssc.sparkContext.setLogLevel("ERROR")

    //监听本地套接字，端口号 9999
    val lines9999: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
    val lines8888: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 8888)

    val map9999: DStream[(String, String)] = lines9999.map((_, "9999"))
    val map8888: DStream[(String, String)] = lines8888.map((_, "8888"))

    val res: DStream[(String, (String, String))] = map9999.join(map8888)


   res.print()



    ssc.start()
    ssc.awaitTermination()


  }
}
